package net.wlm.realtime.funcation;

import ch.hsr.geohash.GeoHash;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.utils.DateTimeUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.DecimalFormat;

public class CarProcessAllWindowFunction extends ProcessAllWindowFunction<VehicleData, String, TimeWindow> {

    // 状态管理
    private transient MapState<String, Integer> heatMapState;
    // 格式化工具
    private static final DecimalFormat df = new DecimalFormat("#.00");
    private transient MapState<String, Boolean> vehicleProcessedState;
    @Override
    public void open(Configuration parameters) {
        // 初始化 MapState（键: VIN码, 值: Boolean标记是否已处理）
        MapStateDescriptor<String, Boolean> descriptor =
                new MapStateDescriptor<>(
                        "vehicleProcessedState",  // 状态名称
                        String.class,            // Key类型（VIN码）
                        Boolean.class            // Value类型
                );
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1)) // 状态保留 1 天
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        descriptor.enableTimeToLive(ttlConfig); // 启用 TTL
        vehicleProcessedState = getRuntimeContext().getMapState(descriptor);
    }
    @Override
    public void process(ProcessAllWindowFunction<VehicleData, String, TimeWindow>.Context ctx,
                        Iterable<VehicleData> iter, Collector<String> out) throws Exception {
        int cnt = 0;
        int i = 0;
        double sum = 0;
        String avg = null;
        for (VehicleData data : iter) {
            // 计算车辆
            if(!vehicleProcessedState.contains(data.getVin())){
                // 统计车辆vin,防止重复统计
                vehicleProcessedState.put(data.getVin(), true);
                cnt++;
            }
            i++;
            sum+=data.getBatteryLevel();
            avg = df.format(sum/i);
        }
        // 输出
        String startTime = DateTimeUtil.convertLongToString(ctx.window().getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String endTime = DateTimeUtil.convertLongToString(ctx.window().getEnd(), DateTimeUtil.DATE_TIME_FORMAT);
        out.collect(startTime+","+endTime+","+cnt+","+avg);
    }
}